package org.example.iotdb.flink.source;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import org.example.iotdb.flink.entity.Message;

import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;
import java.util.Random;


/**
 * @Auther: kevin
 * @Description:
 * @Company: 上海博般数据技术有限公司
 * @Version: 1.0.0
 * @Date: Created in 16:08 2021/8/19
 * @ProjectName: iotdb-learn
 */
public class AutoGenerateDataSource extends RichSourceFunction<Tuple2<String, Message>> {

    private boolean isRunning = true;

    private List<String> pathPrefix = Arrays.asList(
            "G134021710.T33807.C1208490873.M2012000018372095",
            "G134021710.T33809.C1208490886.M2012000018600028",
            "G134021710.T33830.C1208490903.M2012000018372073",
            "G134021710.T51288.C1215120936.M2012000018372079",
            "G134021702.T54800.C1214530055.M2012000018372105",
            "G134021709.T33820.C1215088715.M2012000018600091",
            "G134021709.T54113.C1215089691.M2012000018600035",
            "G134021709.T39868.C1215089356.M2012000018372115",
            "G134021705.T214098099.C1200145153.M8000000048280373",
            "G134021709.T39856.C1215089330.M2012000018600042",

            "G134021709.T39867.C1215089343.M2012000018600012",
            "G134021715.T36189.C1203038106.M2012000018372104",
            "G134021715.T53787.C1213826438.M2012000018372019",
            "G134021706.T2013000172725686.C2209359691.M8000000042181713",
            "G134021705.T2013000182322948.C2209359486.M8000000042181702",
            "G134021713.T2013000182314786.C2209359880.M8000000042181497",
            "G134021704.T39761.C1214419868.M2012000020804010",
            "G134021704.T214099899.C1214419578.M8000000048282854",
            "G134021715.T59492.C1213826500.M8000000042181941",
            "G134021713.T214099531.C1200020876.M8000000048284243",

            "G134021705.T36497.C1200145889.M2012000019779025",
            "G134021702.T257122110.C2208163684.M8000000048279760",
            "G134021706.T214098009.C1200050651.M8000000048282725",
            "G134021702.T348472158.C2208640688.M8000000049212779",
            "G134021708.T214099021.C1200050693.M8000000048279222",
            "G134021711.T214098117.C1200145238.M8000000048279206",
            "G134021711.T214098119.C1200145515.M8000000048280258",
            "G134021715.T2014000192123279.C2210003680.M8000000042181424",
            "G134021715.T2014000192016302.C2210007288.M8000000042181218",
            "G134021715.T2014000192019005.C2210007493.M8000000042182265",

            "G134021713.T33697.C1215055739.M8000000042845184",
            "G134021702.T231750825.C1202416460.M8000000049191239",
            "G134021703.T2017000276367030.C1207211969.M8000000046965800",
            "G134021704.T214099913.C1204484939.M8000000044487108",
            "G134021702.T231751115.C2210132306.M2013000058662089",
            "G134021703.T2017000276367030.C1204205446.M8000000046965366",
            "G134021713.T36240.C1215055973.M8000000042181972",
            "G134021704.T2017000276361022.C1214419480.M8000000048281691",
            "G134021705.T214098907.C1200144932.M8000000047039200",
            "G134021715.T214099607.C1203038005.M8000000043362194",

            "G134021704.T2018000282457013.C1214419477.M8000000048283264",
            "G134021715.T214099601.C2201636011.M8000000042842311",
            "G134021708.T214098985.C1208408432.M2011000013963249",
            "G134021710.T214099581.C1208491108.M8000000048210078",
            "G134021704.T2014000195382612.C2210259894.M100000259515228",
            "G134021705.T214098899.C1200144903.M8000000048209431",
            "G134021705.T214098905.C1200144958.M8000000048279991",
            "G134021705.T255829248.C1200144961.M8000000048280581",
            "G134021706.T214099029.C1200050299.M8000000043363521",
            "G134021715.T2014000187699078.C2209679100.M8000000043190369"
    );

    @Override
    public void run(SourceContext<Tuple2<String, Message>> sourceContext) throws Exception {
        Random random = new Random();
        while (isRunning) {
            long time = System.currentTimeMillis();
            for (String prefix : pathPrefix) {
                Message message = new Message();
                message.setPath(prefix);
                message.setTime(time);
                message.setP(dataFormat(random.nextDouble() * 100, 2));
                message.setQ(dataFormat(random.nextDouble() * 100, 2));
                message.setI(dataFormat(random.nextDouble() * 100, 2));
                message.setU(dataFormat(random.nextDouble() * 100, 2));
                sourceContext.collect(new Tuple2<>(prefix.split("\\.")[0], message));
            }
            Thread.sleep(50);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    public double dataFormat(double value, int decimals) {
        if (decimals == 0) {
            decimals = 4;
        }
        BigDecimal decimal = new BigDecimal(value);
        return decimal.setScale(decimals, BigDecimal.ROUND_HALF_UP).doubleValue();
    }

}
